package com.example.kafkademo.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

public class KafkaConsumerDemo {

    private static int i = 0;

    public static void main(String[] args) throws InterruptedException {
        Properties properties = new Properties();
//        properties.put("bootstrap.servers", "10.42.25.15:9092,10.42.25.12:9092,10.42.25.13:9092");
        properties.put("bootstrap.servers", "localhost:9092");
        properties.put("group.id", "12334");
        properties.put("enable.auto.commit", "true");
        properties.put("auto.commit.interval.ms", "100");
        properties.put("auto.offset.reset", "earliest");
        properties.put("session.timeout.ms", "30000");
        properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
//        properties.put("max.poll.records", "2000000");    //默认500，   修改后增加到3000
//        properties.put("max.partition.fetch.bytes", "990000000");    //默认500，   修改后增加到  15 万
//        properties.put("fetch.max.bytes", "990000000");    //默认500，   修改后增加到  25 万

        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
        kafkaConsumer.subscribe(Arrays.asList("test-a"));


        Long start = System.nanoTime();
        try{
            while (true) {
                ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(100));

                if(records.count() > 1){
                    System.out.println(records.count());
                    for(ConsumerRecord consumerRecord : records){
                        System.out.println("^^^^^^^" + consumerRecord.partition() + "=====" + consumerRecord.topic().length());
                    }
                    System.out.println("^^^^^^^^^^^^^^^^^^^^^" + records.count());
                    System.out.println("=======:分区号" + records.partitions());
                }
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("offset = %d, value = %s, parition = %d", record.offset(), record.value(),record.partition());
                    i++;
                    Long end = System.nanoTime();
                    System.out.println("================="+ (end - start) );
                }
            }
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            System.out.println("消费消息总数:" + i);
        }
        System.out.println("消费消息总数:" + i);



    }


}
